//package com.atguigu.day00;
//
//import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.api.common.functions.AggregateFunction;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
//import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
//import org.apache.flink.streaming.api.windowing.time.Time;
//import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
//
//import java.sql.Timestamp;
//
//public class TopN {
//    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//
//        env.readTextFile("D:\\WebSite\\flink\\flink\\java_flink\\src\\main\\resources\\UserBehavior.csv")
//                .map(x -> {
//                    String[] arr = x.split(",");
//                    return new UserView(arr[0], arr[1], arr[2], arr[3], Long.parseLong(arr[4]));
//                })
//                .filter(x->x.behavior.equals("pv"))
//                .assignTimestampsAndWatermarks(WatermarkStrategy.<UserView>forMonotonousTimestamps()
//                .withTimestampAssigner((SerializableTimestampAssigner<UserView>) (element, recordTimestamp) -> element.timestamp))
//                .keyBy(x->x.itemId)
//                .window(SlidingEventTimeWindows.of(Time.hours(1),Time.minutes(5)))
//                .aggregate(new AggregateFunction<UserView, Long, Long>() {
//                    @Override
//                    public Long createAccumulator() {
//                        return 0L;
//                    }
//
//                    @Override
//                    public Long add(UserView value, Long accumulator) {
//                        return accumulator + 1;
//                    }
//
//                    @Override
//                    public Long getResult(Long accumulator) {
//                        return accumulator;
//                    }
//
//                    @Override
//                    public Long merge(Long a, Long b) {
//                        return null;
//                    }
//                }, new ProcessWindowFunction<Long, ItemViewCount, String, TimeWindow>() {
//                })
//
//        env.execute();
//    }
//
//    public static class UserView {
//        public String userId;
//        public String itemId;
//        public String categoryId;
//        public String behavior;
//        public Long timestamp;
//
//        @Override
//        public String toString() {
//            return "UserView{" +
//                    "userId='" + userId + '\'' +
//                    ", itemId='" + itemId + '\'' +
//                    ", categoryId='" + categoryId + '\'' +
//                    ", behavior='" + behavior + '\'' +
//                    ", timestamp=" + new Timestamp(timestamp) +
//                    '}';
//        }
//
//        public UserView() {
//        }
//
//        public UserView(String userId, String itemId, String categoryId, String behavior, Long timestamp) {
//            this.userId = userId;
//            this.itemId = itemId;
//            this.categoryId = categoryId;
//            this.behavior = behavior;
//            this.timestamp = timestamp * 1000L;
//        }
//    }
//    public static class ItemViewCount{
//        public String itemId;
//        public Long windowStart;
//        public Long windowEnd;
//        public Long count;
//
//        @Override
//        public String toString() {
//            return "ItemViewCount{" +
//                    "itemId='" + itemId + '\'' +
//                    ", windowStart=" + windowStart +
//                    ", windowEnd=" + windowEnd +
//                    ", count=" + count +
//                    '}';
//        }
//
//        public ItemViewCount() {
//        }
//
//        public ItemViewCount(String itemId, Long windowStart, Long windowEnd, Long count) {
//            this.itemId = itemId;
//            this.windowStart = windowStart;
//            this.windowEnd = windowEnd;
//            this.count = count;
//        }
//    }
//}
